package com.cn.daimajiangxin.flink;

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.connector.file.src.FileSource;
import org.apache.flink.connector.file.src.reader.StreamFormat;
import org.apache.flink.connector.file.src.reader.TextLineInputFormat;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.Arrays;

public class BatchWordCount {
    public static void main(String[] args) throws Exception {
        // 创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 明确设置为批处理模式
        env.setRuntimeMode(RuntimeExecutionMode.BATCH);

        // 从文件读取数据（有界数据源）
        String inputPath = "path\\flink-learning\\data\\input.txt";
        // 1. 创建文件源构建器
        Path filePath = new Path(inputPath);

        // 2. 配置文件读取格式
        StreamFormat<String> format =new TextLineInputFormat("UTF-8");

        // 3. 构建 FileSource
        FileSource<String> fileSource = FileSource
                .forRecordStreamFormat(format, filePath)
                .build();
        // 4. 添加 Watermark 策略（批处理中可使用默认策略）
        WatermarkStrategy<String> watermarkStrategy = WatermarkStrategy
                .<String>forMonotonousTimestamps()
                .withIdleness(Duration.ofSeconds(10));

        DataStream<String> text = env.fromSource(fileSource,watermarkStrategy,"FileSource");

        // 数据处理逻辑
        DataStream<Tuple2<String, Integer>> counts = text
                .flatMap(new Tokenizer())
                .keyBy(value -> value.f0)
                .sum(1);

        // 输出结果
        counts.print();

        // 执行作业
        env.execute("Batch Word Count");
    }

    public static final class Tokenizer implements FlatMapFunction<String, Tuple2<String, Integer>> {
        private static final long serialVersionUID = 1L;
        @Override
        public void flatMap(String value, Collector<Tuple2<String, Integer>> out) {
            // 分词并为每个单词生成(word, 1)的元组
            Arrays.stream(value.toLowerCase().split("\\W+"))
                    .filter(word -> word.length() > 0)
                    .forEach(word -> out.collect(new Tuple2<>(word, 1)));
        }
    }
}
